package platon

import (
	"fmt"
	"github.com/Shopify/sarama"
)

type KafkaConsumer struct {
	Node  string
	Topic string
	//Message string
	MessageQueue chan string
}

func NewKafkaConsumer(node, topic string) *KafkaConsumer {
	return &KafkaConsumer{
		Node:         node,
		Topic:        topic,
		MessageQueue: make(chan string, 10),
	}
}

func (this KafkaConsumer) Consume() {
	consumer, err := sarama.NewConsumer([]string{this.Node}, nil)
	if err != nil {
		fmt.Printf("kafka connnet failed, error[%v]", err.Error())
		return
	}
	defer consumer.Close()

	partitions, err := consumer.Partitions(this.Topic)
	if err != nil {
		fmt.Printf("get topic failed, error[%v]", err.Error())
		return
	}
	for _, p := range partitions {
		partitionConsumer, err := consumer.ConsumePartition(this.Topic, p, sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("get partition consumer failed, error[%v]", err.Error())
			continue
		}

		for message := range partitionConsumer.Messages() {
			fmt.Printf("message:[%v], key:[%v], offset:[%v]\n", string(message.Value), string(message.Key), string(message.Offset))
			HandleMessage(message.Value)
		}
	}
}
